package com.zhaosc.rx.example;

import java.util.List;
import java.util.concurrent.TimeUnit;

import org.junit.Test;

import com.google.common.collect.Lists;
import com.zhaosc.rx.utils.SleepUtils;

import io.reactivex.Observable;

public class MergeOrZipExample {

	/**
	 * 如果只是两个被观察者合井，则还可以使用 mergeWith操作符， Observable.merge(odds,evens)等价于
	 * odds.mergeWith(evens)
	 */
	@Test
	public void testMerge() {
		Observable.merge(Observable.just("a", "e", "d"), Observable.just("b", "c", "f")).sorted()
				.subscribe((t) -> System.out.println(t));
	}

	/**
	 * 通过一个函数将多个 Observable 的发射物结合到一起， 基于这个函数的结果为每个结合体 发射单个数据项; zip 只发射与发射数据项最少 的那个
	 * Observable一样多的数据。
	 */
	@Test
	public void testZip() {
		Observable.zip(Observable.just("a", "e", "d"), Observable.just("b", "c", "f"), (t1, t2) -> {
			return t1 + t2;
		}).subscribe(t -> System.out.println(t));

		List<Observable<String>> observables = Lists.newArrayList(Observable.just("a", "e", "d"),
				Observable.just("b", "c"));
		Observable.zip(observables, (t) -> {
			String t1 = "";
			System.out.println("******" + t);
			for (Object t2 : t) {
				t1 = t1 + t2;
			}
			return t1;
		}).subscribe(t -> System.out.println(t));

	}

	/**
	 * 第一个source的最后一个数据进行操作
	 */
	@Test
	public void testCombineLatest() {

		Observable.combineLatest(Observable.just(1, 3, 7), Observable.just(2, 4, 8), (t1, t2) -> {
			return t1 + t2;
		}).subscribe(t -> System.out.println(t));
	}

	/**
	 * join 操作符结合两个 Observable 发射的数据，
	 * 基于时间窗口(针对每条数据特定的原则) 选择待集合的数据项。
	 * 将这些 时间窗口 实现为一些 Observables，
	 * 它们的生命周期从任何一条 Observable 发射的每一条数据开始。
	 * 当这个定义时间窗口的 Observable 发射了一条数据或者完 成时，
	 * 与这条数据关联的窗口也会关闭 。只 要这条数据的窗口是打开的，
	 * 它就继续结合其他 Observable 发射的任何数据项。
	 */
	@Test
	public void testJoin() {
//    join(ObservableSource<? extends TRight>, Function<? super T, ? extends ObservableSource<TLeftEnd>>, Function<? super TRight, ? extends ObservableSource<TRightEnd>>, BiFunction<? super T, ? super TRight, ? extends R>)
//		Observable:源 Observable 需要组合 的 Observable，这里可以称之为目标 Observable。
//		Function: 接收从源 Observable 发射来的数据，井返回一个 Observable，这个 Observable
//		的生命周期决定了源 Observable 发射数据的有效期。
//		Function: 接收目标 Observable 发射的数据，井返回一个 Observable，这个 Observable
//		的生命周期决定了目标 Observable发射数据的有效期。
//		BiFunction: 接收从源 Observable 和目 标 Observable 发射的数据，并将这两个数据组
//		合后返回。
		Observable.just("a", "b", "c").join(Observable.just("d", "e", "f"), (t) -> {
			return Observable.just(t).delay(1, TimeUnit.MICROSECONDS);
		}, (t) -> {
			return Observable.just(t).delay(2, TimeUnit.MICROSECONDS);
		}, (t1, t2) -> {
			return t1 + t2;
		}).subscribe(t -> System.out.println(t));
		SleepUtils.sleep(100000L);
	}
	
	@Test
	public void testStartWith() {
		Observable.just("a", "b", "c").startWith("hahah")
		.subscribe((t)->System.out.println(t));
	}
	
	@Test
	public void teststartWithArray() {
		Observable.just("a", "b", "c")
		.startWithArray("hahah","oooo")
		//.startWith("TTTT")优先级比较高
		.subscribe((t)->System.out.println(t));
	}
	
}
